package org.bigwinner.flinkLearning.Sinks;

import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
import org.bigwinner.flinkLearning.Constants.ConstantsValue;
import org.bigwinner.flinkLearning.Message.MysqlMessage;
import org.bigwinner.flinkLearning.utils.JdbcUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: IT大狮兄
 * @date: 2021/7/30 上午10:35
 * @version: 1.0.0
 * @description: mysql两阶段提交
 */
public class MySqlTwoPhaseCommitSink extends TwoPhaseCommitSinkFunction<MysqlMessage, String, Void> {
    private static final Logger LOGGER = LoggerFactory.getLogger(MySqlTwoPhaseCommitSink.class);
    private String mysqlTb = ConstantsValue.MYSQL_TABLE;
    private ConcurrentHashMap<String, Connection> connections;
    public MySqlTwoPhaseCommitSink() {
        super(StringSerializer.INSTANCE, VoidSerializer.INSTANCE);
        connections = new ConcurrentHashMap<>();
    }

    @Override
    protected void invoke(String transactionId, MysqlMessage mysqlMessage, Context context) throws Exception {
        LOGGER.info(mysqlTb + " start invoke...");
        String sql = "insert into " + mysqlTb + "(icid, telnum) values (?,?)";
        PreparedStatement insertStmt = this.connections.get(transactionId).prepareStatement(sql);
        insertStmt.setString(1, mysqlMessage.getIcid());
        insertStmt.setString(2, mysqlMessage.getTelnum());
        insertStmt.execute();
    }

    @Override
    protected String beginTransaction() throws Exception {
        LOGGER.info(mysqlTb + " start beginTransaction...");
        Connection connection = JdbcUtils.getConnection();
        String transId = UUID.randomUUID().toString();
        System.out.println(transId + connection);
        this.connections.put(transId, connection);
        return transId;
    }

    @Override
    protected void preCommit(String transactionId) throws Exception {
        LOGGER.info(mysqlTb + " start preCommit...");
    }

    @Override
    protected void commit(String transactionId) {
        LOGGER.info(mysqlTb + " start commit...");
        JdbcUtils.commit(this.connections.remove(transactionId));
    }

    @Override
    protected void abort(String transactionId) {
        LOGGER.error(mysqlTb + " start abort...");
        JdbcUtils.rollback(this.connections.remove(transactionId));
    }
}
